/*
 * brin_xlog.c
 *		BRIN 索引的 XLog 重放例程
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/access/brin/brin_xlog.c
 */
#include "postgres.h"

#include "access/brin_page.h"
#include "access/brin_pageops.h"
#include "access/brin_xlog.h"
#include "access/bufmask.h"
#include "access/xlogutils.h"


/*
 * xlog 重放例程
 */
static void fc_brin_xlog_createidx(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_brin_createidx *fc_xlrec = (xl_brin_createidx *) XLogRecGetData(fc_record);
	Buffer		fc_buf;
	Page		fc_page;

	/* 创建索引的元页面 */
	fc_buf = XLogInitBufferForRedo(fc_record, 0);
	Assert(BufferIsValid(fc_buf));
	fc_page = (Page) BufferGetPage(fc_buf);
	brin_metapage_init(fc_page, fc_xlrec->pagesPerRange, fc_xlrec->version);
	PageSetLSN(fc_page, fc_lsn);
	MarkBufferDirty(fc_buf);
	UnlockReleaseBuffer(fc_buf);
}

/*
 * 插入或更新的公共部分。插入新的元组并更新
 * revmap。
 */
static void fc_brin_xlog_insert_update(XLogReaderState *fc_record,
						xl_brin_insert *fc_xlrec)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	Buffer		fc_buffer;
	BlockNumber fc_regpgno;
	Page		fc_page;
	XLogRedoAction fc_action;

	/*
	 * 如果我们在页面上插入了第一个也是唯一的元组，重新初始化页面。
	 */
	if (XLogRecGetInfo(fc_record) & XLOG_BRIN_INIT_PAGE)
	{
		fc_buffer = XLogInitBufferForRedo(fc_record, 0);
		fc_page = BufferGetPage(fc_buffer);
		brin_page_init(fc_page, BRIN_PAGETYPE_REGULAR);
		fc_action = BLK_NEEDS_REDO;
	}
	else
	{
		fc_action = XLogReadBufferForRedo(fc_record, 0, &fc_buffer);
	}

	/* 需要此页面的 blkno 存储在 revmap 中 */
	fc_regpgno = BufferGetBlockNumber(fc_buffer);

	/* 将索引项插入页面 */
	if (fc_action == BLK_NEEDS_REDO)
	{
		OffsetNumber fc_offnum;
		BrinTuple  *fc_tuple;
		Size		fc_tuplen;

		fc_tuple = (BrinTuple *) XLogRecGetBlockData(fc_record, 0, &fc_tuplen);

		Assert(fc_tuple->bt_blkno == fc_xlrec->heapBlk);

		fc_page = (Page) BufferGetPage(fc_buffer);
		fc_offnum = fc_xlrec->offnum;
		if (PageGetMaxOffsetNumber(fc_page) + 1 < fc_offnum)
			elog(PANIC, "brin_xlog_insert_update: invalid max offset number");

		fc_offnum = PageAddItem(fc_page, (Item) fc_tuple, fc_tuplen, fc_offnum, true, false);
		if (fc_offnum == InvalidOffsetNumber)
			elog(PANIC, "brin_xlog_insert_update: failed to add tuple");

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);

	/* 更新 revmap */
	fc_action = XLogReadBufferForRedo(fc_record, 1, &fc_buffer);
	if (fc_action == BLK_NEEDS_REDO)
	{
		ItemPointerData fc_tid;

		ItemPointerSet(&fc_tid, fc_regpgno, fc_xlrec->offnum);
		fc_page = (Page) BufferGetPage(fc_buffer);

		brinSetHeapBlockItemptr(fc_buffer, fc_xlrec->pagesPerRange, fc_xlrec->heapBlk,
								fc_tid);
		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);

	/* XXX 这里没有 FSM 更新... */
}

/*
 * 重放 BRIN 索引插入
 */
static void fc_brin_xlog_insert(XLogReaderState *fc_record)
{
	xl_brin_insert *fc_xlrec = (xl_brin_insert *) XLogRecGetData(fc_record);

	fc_brin_xlog_insert_update(fc_record, fc_xlrec);
}

/*
 * 重放 BRIN 索引更新
 */
static void fc_brin_xlog_update(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_brin_update *fc_xlrec = (xl_brin_update *) XLogRecGetData(fc_record);
	Buffer		fc_buffer;
	XLogRedoAction fc_action;

	/* 首先移除旧元组 */
	fc_action = XLogReadBufferForRedo(fc_record, 2, &fc_buffer);
	if (fc_action == BLK_NEEDS_REDO)
	{
		Page		fc_page;
		OffsetNumber fc_offnum;

		fc_page = (Page) BufferGetPage(fc_buffer);

		fc_offnum = fc_xlrec->oldOffnum;

		PageIndexTupleDeleteNoCompact(fc_page, fc_offnum);

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}

	/* 然后插入新元组并更新 revmap，像插入时那样。 */
	fc_brin_xlog_insert_update(fc_record, &fc_xlrec->insert);

	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);
}

/*
 * 在单个页面上更新元组。
 */
static void fc_brin_xlog_samepage_update(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_brin_samepage_update *fc_xlrec;
	Buffer		fc_buffer;
	XLogRedoAction fc_action;

	fc_xlrec = (xl_brin_samepage_update *) XLogRecGetData(fc_record);
	fc_action = XLogReadBufferForRedo(fc_record, 0, &fc_buffer);
	if (fc_action == BLK_NEEDS_REDO)
	{
		Size		fc_tuplen;
		BrinTuple  *fc_brintuple;
		Page		fc_page;
		OffsetNumber fc_offnum;

		fc_brintuple = (BrinTuple *) XLogRecGetBlockData(fc_record, 0, &fc_tuplen);

		fc_page = (Page) BufferGetPage(fc_buffer);

		fc_offnum = fc_xlrec->offnum;

		if (!PageIndexTupleOverwrite(fc_page, fc_offnum, (Item) fc_brintuple, fc_tuplen))
			elog(PANIC, "brin_xlog_samepage_update: failed to replace tuple");

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);

	/* XXX 这里没有 FSM 更新... */
}

/*
 * 重放 revmap 页面扩展
 */
static void fc_brin_xlog_revmap_extend(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_brin_revmap_extend *fc_xlrec;
	Buffer		fc_metabuf;
	Buffer		fc_buf;
	Page		fc_page;
	BlockNumber fc_targetBlk;
	XLogRedoAction fc_action;

	fc_xlrec = (xl_brin_revmap_extend *) XLogRecGetData(fc_record);

	XLogRecGetBlockTag(fc_record, 1, NULL, NULL, &fc_targetBlk);
	Assert(fc_xlrec->targetBlk == fc_targetBlk);

	/* 更新元页面 */
	fc_action = XLogReadBufferForRedo(fc_record, 0, &fc_metabuf);
	if (fc_action == BLK_NEEDS_REDO)
	{
		Page		fc_metapg;
		BrinMetaPageData *fc_metadata;

		fc_metapg = BufferGetPage(fc_metabuf);
		fc_metadata = (BrinMetaPageData *) PageGetContents(fc_metapg);

		Assert(fc_metadata->lastRevmapPage == fc_xlrec->targetBlk - 1);
		fc_metadata->lastRevmapPage = fc_xlrec->targetBlk;

		PageSetLSN(fc_metapg, fc_lsn);

		/*
		 * 将 pd_lower 设置在元数据末尾之后。这是必需的，
		 * 因为如果不这样做，元数据将在 xlog.c
		 * 压缩页面时丢失。（我们必须在这里这样做，因为在 v11
		 * 之前的 PG 版本没有正确设置元页面的 pd_lower，因此一个
		 * pg_upgraded 索引可能包含错误的值。）
		 */
		((PageHeader) fc_metapg)->pd_lower =
			((char *) fc_metadata + sizeof(BrinMetaPageData)) - (char *) fc_metapg;

		MarkBufferDirty(fc_metabuf);
	}

	/*
	 * 将目标块重新初始化为 revmap 页面。这里从来没有完整页面
	 * 图像。
	 */

	fc_buf = XLogInitBufferForRedo(fc_record, 1);
	fc_page = (Page) BufferGetPage(fc_buf);
	brin_page_init(fc_page, BRIN_PAGETYPE_REVMAP);

	PageSetLSN(fc_page, fc_lsn);
	MarkBufferDirty(fc_buf);

	UnlockReleaseBuffer(fc_buf);
	if (BufferIsValid(fc_metabuf))
		UnlockReleaseBuffer(fc_metabuf);
}

static void fc_brin_xlog_desummarize_page(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	xl_brin_desummarize *fc_xlrec;
	Buffer		fc_buffer;
	XLogRedoAction fc_action;

	fc_xlrec = (xl_brin_desummarize *) XLogRecGetData(fc_record);

	/* 更新 revmap */
	fc_action = XLogReadBufferForRedo(fc_record, 0, &fc_buffer);
	if (fc_action == BLK_NEEDS_REDO)
	{
		ItemPointerData fc_iptr;

		ItemPointerSetInvalid(&fc_iptr);
		brinSetHeapBlockItemptr(fc_buffer, fc_xlrec->pagesPerRange, fc_xlrec->heapBlk, fc_iptr);

		PageSetLSN(BufferGetPage(fc_buffer), fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);

	/* 从常规页面中移除剩余条目 */
	fc_action = XLogReadBufferForRedo(fc_record, 1, &fc_buffer);
	if (fc_action == BLK_NEEDS_REDO)
	{
		Page		fc_regPg = BufferGetPage(fc_buffer);

		PageIndexTupleDeleteNoCompact(fc_regPg, fc_xlrec->regOffset);

		PageSetLSN(fc_regPg, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);
}

void brin_redo(XLogReaderState *fc_record)
{
	uint8		fc_info = XLogRecGetInfo(fc_record) & ~XLR_INFO_MASK;

	switch (fc_info & XLOG_BRIN_OPMASK)
	{
		case XLOG_BRIN_CREATE_INDEX:
			fc_brin_xlog_createidx(fc_record);
			break;
		case XLOG_BRIN_INSERT:
			fc_brin_xlog_insert(fc_record);
			break;
		case XLOG_BRIN_UPDATE:
			fc_brin_xlog_update(fc_record);
			break;
		case XLOG_BRIN_SAMEPAGE_UPDATE:
			fc_brin_xlog_samepage_update(fc_record);
			break;
		case XLOG_BRIN_REVMAP_EXTEND:
			fc_brin_xlog_revmap_extend(fc_record);
			break;
		case XLOG_BRIN_DESUMMARIZE:
			fc_brin_xlog_desummarize_page(fc_record);
			break;
		default:
			elog(PANIC, "brin_redo: unknown op code %u", fc_info);
	}
}

/*
 * 在进行一致性检查之前屏蔽 BRIN 页面。
 */
void brin_mask(char *fc_pagedata, BlockNumber fc_blkno)
{
	Page		fc_page = (Page) fc_pagedata;
	PageHeader	fc_pagehdr = (PageHeader) fc_page;

	mask_page_lsn_and_checksum(fc_page);

	mask_page_hint_bits(fc_page);

	/*
	 * 常规 brin 页面包含需要屏蔽的未使用空间。
	 * 元页面也是如此，但仅在 pd_lower 似乎已
	 * 正确设置时才进行屏蔽。
	 */
	if (BRIN_IS_REGULAR_PAGE(fc_page) ||
		(BRIN_IS_META_PAGE(fc_page) && fc_pagehdr->pd_lower > SizeOfPageHeaderData))
	{
		mask_unused_space(fc_page);
	}

	/*
	 * BRIN_EVACUATE_PAGE 没有 WAL 日志记录，因为它在恢复中没有用处。
	 * 进行屏蔽。有关详细信息，请参见 brin_start_evacuating_page()。
	 */
	BrinPageFlags(fc_page) &= ~BRIN_EVACUATE_PAGE;
}
